package org.zjvis.datascience.common.util;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.security.KeyManagementException;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import javax.annotation.PostConstruct;
import javax.net.ssl.SSLContext;

import net.zjvis.lab.nebula.backend.lib.algorithm.conf.algorithm.Algorithm;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.conn.ssl.NoopHostnameVerifier;
import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
import org.apache.http.conn.ssl.TrustSelfSignedStrategy;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.*;
import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.client.RequestCallback;
import org.springframework.web.client.RestTemplate;
import net.zjvis.lab.nebula.backend.lib.algorithm.conf.Configuration;
import org.zjvis.datascience.common.model.ApiResult;
import org.zjvis.datascience.common.model.ApiResultCode;
import org.zjvis.datascience.common.vo.JobStatusVO;
import reactor.core.publisher.Flux;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

/**
 * @description rest服务调用工具类
 * @date 2021-11-22
 */
@Component
public class RestTemplateUtil {

//    private final static Logger logger = LoggerFactory.getLogger(RestTemplateUtil.class);

    private RestTemplate restTemplate;

//    private RestTemplate restTemplateForAudit;

    private RestTemplate restTemplateForTimeout;

//    private RestTemplate restTemplateForML;

    private RestTemplate restTemplateForFlask;

//    private String jobServer;

    @Value("${restful.flaskServer.address}")
    private String flaskServer;

    @Value("${restful.insightServer.address}")
    private String insightServer;

    @Value("${upload.folder-path}")
    private String folderPath;

    @Value("${restful.algoServer.address:null}")
    private String algoServer;

    public String getFlaskServer(){
        return this.flaskServer;
    }

    @PostConstruct
    public void init() throws KeyStoreException, NoSuchAlgorithmException, KeyManagementException {
        SSLContext sslContext = org.apache.http.ssl.SSLContexts.custom()
                .loadTrustMaterial(null, new TrustSelfSignedStrategy())
                .build();

        SSLConnectionSocketFactory csf = new SSLConnectionSocketFactory(sslContext, NoopHostnameVerifier.INSTANCE);

        CloseableHttpClient httpClient = HttpClients.custom()
                .setSSLSocketFactory(csf)
                .setMaxConnTotal(200)
                .setMaxConnPerRoute(100)
                .build();

        HttpComponentsClientHttpRequestFactory requestFactory = new HttpComponentsClientHttpRequestFactory();
        requestFactory.setHttpClient(httpClient);

        this.restTemplate = new RestTemplate(requestFactory);

//        initAuditRestTemplate();

        initTimeoutTemplate(15 * 1000);

//        initMLRestTemplate();

        initFlaskRestTemplate();

//        setLocalJobServer();
    }

//    private void setLocalJobServer() {
//        this.jobServer = "https://localhost:" + port;
//        logger.info("Using jobserver at {}", this.jobServer);
//    }

//    private String getJobServer() {
//        return this.jobServer;
//    }
//
//    private void initAuditRestTemplate() {
//        HttpComponentsClientHttpRequestFactory clientHttpRequestFactory = new HttpComponentsClientHttpRequestFactory();
//        clientHttpRequestFactory.setConnectionRequestTimeout(2000);
//        clientHttpRequestFactory.setReadTimeout(2000);
//        clientHttpRequestFactory.setConnectTimeout(2000);
//        restTemplateForAudit = new RestTemplate(clientHttpRequestFactory);
//    }
//
//    private void initMLRestTemplate() {
//        HttpComponentsClientHttpRequestFactory clientHttpRequestFactory = new HttpComponentsClientHttpRequestFactory();
//        clientHttpRequestFactory.setConnectionRequestTimeout(20000);
//        clientHttpRequestFactory.setReadTimeout(20000);
//        clientHttpRequestFactory.setConnectTimeout(20000);
//        restTemplateForML = new RestTemplate(clientHttpRequestFactory);
//    }

    private void initFlaskRestTemplate() {
        HttpComponentsClientHttpRequestFactory clientHttpRequestFactory = new HttpComponentsClientHttpRequestFactory();
        clientHttpRequestFactory.setConnectionRequestTimeout(800000);
        clientHttpRequestFactory.setReadTimeout(800000);
        clientHttpRequestFactory.setConnectTimeout(800000);
        restTemplateForFlask = new RestTemplate(clientHttpRequestFactory);
    }

    private void initTimeoutTemplate(Integer readTimeout) {
        HttpComponentsClientHttpRequestFactory clientHttpRequestFactory = new HttpComponentsClientHttpRequestFactory();
        clientHttpRequestFactory.setConnectTimeout(2000);
        clientHttpRequestFactory.setReadTimeout(readTimeout);
        restTemplateForTimeout = new RestTemplate(clientHttpRequestFactory);

    }

    /**
     * 调用jobserver提交任务
     *
     * @param appArgs 命令行参数
     * @return applicationId, 为空就是提交失败
     */
//    public String submitJob(String appArgs) {
//        String url = getJobServer() + "/job/submitJob?appArgs=" + appArgs;
//        JSONObject reqBody = new JSONObject();
//        reqBody.put("appArgs", appArgs);
//        HttpEntity<JSONObject> req = new HttpEntity<>(reqBody);
//        ResponseEntity<ApiResult> entity = restTemplate
//                .exchange(url, HttpMethod.POST, req, ApiResult.class);
//        ApiResult body = entity.getBody();
//        return (String) body.getResult();
//    }

    /**
//     * 调用jobserver提交任务
//     *
//     * @param appArgs 命令行参数
//     * @return applicationId, 为空就是提交失败
//     */
//    public ApiResult submitModelJob(String appArgs) {
//        String url = getJobServer() + "/job/submitMLJob?appArgs=" + appArgs;
//
//        JSONObject reqBody = new JSONObject();
//        reqBody.put("appArgs", appArgs);
//        HttpEntity<JSONObject> req = new HttpEntity<>(reqBody);
//        ResponseEntity<ApiResult> entity = restTemplate
//                .exchange(url, HttpMethod.POST, req, ApiResult.class);
//        return entity.getBody();
//    }


    /**
     * @param appArgs
     * @param appResourcePath
     * @return
     */
    public String submitPySparkJob(String appArgs, String appResourcePath) throws IOException {
//        String url = String.format("%s/job/submitPySparkJob", getJobServer());
//        JSONObject reqBody = new JSONObject();
//        reqBody.put("appArgs", appArgs);
//        reqBody.put("appResourcePath", appResourcePath);
//        HttpEntity<JSONObject> req = new HttpEntity<>(reqBody);
//
//        ResponseEntity<ApiResult> entity = restTemplateForTimeout
//                .exchange(url, HttpMethod.POST, req, ApiResult.class);
//
//        ApiResult body = entity.getBody();
//        return (String) body.getResult();
        return "";
    }


//    /**
//     * @param params
//     * @return
//     */
//    public String submitMLFlask(JSONObject params) throws IOException {
//        String aPath = params.getString("aPath");
//        String url = String.format("%s/%s", flaskServer, aPath);
//        HttpHeaders headers = new HttpHeaders();
//        headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
//        MultiValueMap<String, String> map = new LinkedMultiValueMap<String, String>();
//        for (String keyStr : params.keySet()) {
//            if (!keyStr.equals("aPath")) {
//                map.add(keyStr, params.getString(keyStr));
//            }
//        }
//        HttpEntity<MultiValueMap<String, String>> request =
//                new HttpEntity<MultiValueMap<String, String>>(map, headers);
//        String result = restTemplateForFlask.postForObject(url, request, String.class);
//        return result;
//    }
    /**
     * @param params
     * @return
     */
    public String submitFlaskJob(JSONObject params) throws IOException {
        String aPath = params.getString("apiPath");
        String url = String.format("%s/%s", flaskServer, aPath);
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
        MultiValueMap<String, String> map = new LinkedMultiValueMap<String, String>();
        for (String keyStr : params.keySet()) {
            if (!keyStr.equals("apiPath")) {
                map.add(keyStr, params.getString(keyStr));
            }
        }
        HttpEntity<MultiValueMap<String, String>> request =
                new HttpEntity<MultiValueMap<String, String>>(map, headers);
        String result = restTemplateForFlask.postForObject(url, request, String.class);
        return result;
    }

    public String submitFlaskJobWebClient(JSONObject params){
//        String urlTag = "modelOnline/";
        String urlSuffix = params.getString("apiPath");
        String url = String.format("%s/%s", flaskServer, urlSuffix);
        if (StringUtils.isNotEmpty(flaskServer)) {
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
            MultiValueMap<String, String> formDataMap = new LinkedMultiValueMap<String, String>();
            for (String keyStr : params.keySet()) {
                formDataMap.add(keyStr, params.getString(keyStr));
            }
            WebClient webClient = WebClient.create();
            String res = "";
            Mono<String> mono = webClient.post().uri(url)
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(BodyInserters.fromFormData(formDataMap))
                    .retrieve().bodyToMono(String.class);
            res = mono.block();
            return res;
        }else {
            return "缺少flask-server模型库的配置信息";
        }
    }


    /**
     * 写pyspark脚本到指定hdfs路径
     *
     * @param appResourcePath 路径名
     * @param scriptBody      脚本内容
     * @return
     */
//    public boolean writePyScriptToHDFS(String appResourcePath, String scriptBody) {
//        String url = String.format("%s/job/writePyScriptToHDFS", getJobServer());
//        JSONObject reqBody = new JSONObject();
//        reqBody.put("scriptBody", scriptBody);
//        reqBody.put("appResourcePath", appResourcePath);
//        HttpEntity<JSONObject> req = new HttpEntity<>(reqBody);
//        ResponseEntity<ApiResult> entity = restTemplate
//                .exchange(url, HttpMethod.POST, req, ApiResult.class);
//        ApiResult body = entity.getBody();
//        return body.getCode() == ApiResultCode.SUCCESS.getCode();
//    }

    /**
     * 查询spark任务的执行状态
     *
     * @param applicationId
     * @return
     */
    public JobStatusVO queryJobStatus(String applicationId, String directory) {
//        String url = getJobServer() + "/job/queryJobStatus";
//        JSONObject json = new JSONObject();
//        json.put("applicationId", applicationId);
//        HttpEntity<JSONObject> req = new HttpEntity<>(json);
//        ResponseEntity<ApiResult> response = restTemplate
//                .exchange(url, HttpMethod.POST, req, ApiResult.class);
//        ApiResult<String> body = response.getBody();
//        if (body.getCode() != ApiResultCode.SUCCESS.getCode()) {
//            return new JobStatusVO();
//        }
//        String jsonString = body.getResult();
//        return JSONObject.parseObject(jsonString, JobStatusVO.class);
        return new JobStatusVO();
    }


    public String invokeAlgo(String configurationJsonStr) {
        if (StringUtils.isNotEmpty(algoServer)) {
            HttpEntity<String> req = new HttpEntity<>(configurationJsonStr);
            ResponseEntity<ApiResult> response = restTemplate
                    .exchange(algoServer, HttpMethod.POST, req, ApiResult.class);
            ApiResult<String> body = response.getBody();
            if (body.getCode() != HttpStatus.OK.value()) {
                return body.getMessage();
            }
            return ObjectUtil.isNotNull(body.getResult()) ? body.getResult() : "";
        }
        return "";
    }

    /**
     * 杀死提交的spark任务
     *
     * @param applicationId
     * @return
     */
    public boolean killJob(String applicationId) {
//        String url = getJobServer() + "/job/killJob";
//        JobStatusVO reqBody = new JobStatusVO();
//        reqBody.setId(applicationId);
//        HttpEntity<JobStatusVO> req = new HttpEntity<>(reqBody);
//        ResponseEntity<ApiResult> response = restTemplate
//                .exchange(url, HttpMethod.POST, req, ApiResult.class);
//        ApiResult body = response.getBody();
//        if (null == body || body.getCode() != ApiResultCode.SUCCESS.getCode()) {
//            return false;
//        }
        return true;
    }

    /**
     * 调jobServer提交任务
     *
     * @param appArgs
     * @return
     */
//    public String runJob(String appArgs) {
//        String url = getJobServer() + "/job/runJob";
//        JSONObject reqBody = new JSONObject();
//        reqBody.put("appArgs", appArgs);
//        HttpEntity<JSONObject> req = new HttpEntity<>(reqBody);
//        logger.info("RestTemplateUtil.runJob(), before send request, req={}, timestamp={}", req,
//                System.currentTimeMillis());
//        ResponseEntity<ApiResult> response;
//        try {
//            //大概耗时2分半  150s 才能拿到结果
//            response = restTemplate.exchange(url, HttpMethod.POST, req, ApiResult.class);
//        } catch (RestClientException e) {
//            e.printStackTrace();
//            return "";
//        }
//        logger.info("RestTemplateUtil.runJob(), received response, response={}, timestamp={}",
//                response, System.currentTimeMillis());
//        ApiResult<String> body = response.getBody();
//        if (body.getCode() != ApiResultCode.SUCCESS.getCode()) {
//            return "";
//        }
//        return body.getResult();
//    }

    //  /**
//   * 发送dataX配置文件到dataX服务器
//   */
//  public String sendDataXConfig(String fileName, String json) {
//    String url = dataXServer + "/upload";
//    DataXDTO dataXBean = new DataXDTO(fileName, json);
//    HttpEntity<DataXDTO> req = new HttpEntity<>(dataXBean);
//    ResponseEntity<String> response = restTemplate
//        .exchange(url, HttpMethod.POST, req, String.class);
//    return response.getBody();
//  }
//
//  /**
//   * 执行dataX任务
//   */
//  public String executeDataXJob(String fileName) {
//    String url = dataXServer + "/execute?fileName=" + fileName;
//    HttpEntity<String> req = new HttpEntity<>(fileName);
//    ResponseEntity<String> response = restTemplate.exchange(url, HttpMethod.GET, req, String.class);
//    return response.getBody();
//  }
    public ApiResult submitMOJob(JSONObject params, String urlSuffix){
        String urlTag = "modelOnline/";
        String url = String.format("%s/%s", insightServer, urlTag + urlSuffix);
        if (StringUtils.isNotEmpty(insightServer)) {
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
            MultiValueMap<String, String> formDataMap = new LinkedMultiValueMap<String, String>();
            for (String keyStr : params.keySet()) {
                formDataMap.add(keyStr, params.getString(keyStr));
            }
            WebClient webClient = WebClient.create();
            ApiResult res = new ApiResult();
            Flux<ApiResult> mono = webClient.post().uri(url)
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(BodyInserters.fromValue(params))
                    .retrieve().bodyToFlux(ApiResult.class);
            res = mono.blockFirst();
            return res;
        }else {
            return ApiResult.valueOf(ApiResultCode.MISSING_CONFIG_INFO, null, "缺少在线模型库的配置信息");
        }
    }

    public String recStreaming(Long id) {

        String url = insightServer + "/modelOnline/install/" + id;
        // 文件保存的本地路径
        String targetDirPath = folderPath;
        File targetDirFile = new File(targetDirPath);

        if (!targetDirFile.exists()){
            targetDirFile.mkdirs();
        }

        String finalTargetPath = targetDirPath + "mo_" + id + ".zip";
        if (id == 0L){
            finalTargetPath = targetDirPath + "script.zip";
        }
        File finalTargetFile = new File(finalTargetPath);

        if (finalTargetFile.exists()){
            return finalTargetPath;
        }
        //定义请求头的接收类型
        RequestCallback requestCallback = request -> request.getHeaders()
                .setAccept(Arrays.asList(MediaType.APPLICATION_OCTET_STREAM, MediaType.ALL));
        //对响应进行流式处理而不是将其全部加载到内存中
        String finalTargetPath1 = finalTargetPath;
        restTemplate.execute(url, HttpMethod.GET, requestCallback, clientHttpResponse -> {
            Files.copy(clientHttpResponse.getBody(), Paths.get(finalTargetPath1));
            return finalTargetPath1;
        });
        return finalTargetPath;
    }

    public String installFromInsight(Long id){
        String aPath = "install?moid=" + id;
        String url = String.format("%s/%s", flaskServer, aPath);
        String result = restTemplateForFlask.getForObject(url, String.class);
        return result;
    }

    public JSONArray getSetParams(Long id){
        String aPath = "getSetParams?moid=" + id;
        String url = String.format("%s/%s", flaskServer, aPath);
        String result = restTemplateForFlask.getForObject(url, String.class);
        return JSONArray.parseArray(result);
    }


    public JSONArray download(Long id){
        String aPath = "getSetParams?moid=" + id;
        String url = String.format("%s/%s", flaskServer, aPath);
        String result = restTemplateForFlask.getForObject(url, String.class);
        return JSONArray.parseArray(result);
    }

    public String get(String url){
        String result = restTemplateForFlask.getForObject(url, String.class);
        return result;
    }

    public <T> ResponseEntity<T> KGraphPostRequest(Map<String, Object> requestParamMap, Class<T> requestType) {
        HttpEntity<Map<String, Object>> request = new HttpEntity<>(requestParamMap);
        return restTemplate.exchange("", HttpMethod.POST, request, requestType);
    }
}
